<?php


namespace App\Implement;


use Hyperf\AsyncQueue\Driver\DriverInterface;
use Hyperf\AsyncQueue\Exception\InvalidQueueException;
use Hyperf\AsyncQueue\JobInterface;
use Hyperf\AsyncQueue\JobMessage;

class RedisDriver extends \Hyperf\AsyncQueue\Driver\RedisDriver implements DriverInterface
{
    //protected $lengthCheckCount = 1;

    /**
     * Notes: 延时任务还有多久执行
     * User: 陈朋
     * DateTime: 2021/7/22 13:26
     * @param JobInterface $job
     * @return int
     */
    public function ttl(JobInterface $job): int
    {
        $message = make(JobMessage::class, [$job]);
        $time = $this->redis->zScore($this->channel->getDelayed(), $this->packer->pack($message));
        if ($time === false) {
            return -1;
        }
        return (int)$time - time();
    }

    /**
     * Notes:重写底层队列reload方法 支持在处理中的任务重载 （主要是hyperf重启后 正在处理的任务会停止 并被转移到timeout）
     * User: 陈朋
     * DateTime: 2021/10/30 15:59
     * @param string|null $queue
     * @return int
     */
    public function reload(string $queue = null): int
    {
        $channel = $this->channel->getFailed();
        if ($queue) {
            if (!in_array($queue, ['timeout', 'failed', 'reserved'])) {
                throw new InvalidQueueException(sprintf('Queue %s is not supported.', $queue));
            }

            $channel = $this->channel->get($queue);
        }

        $num = 0;
        //处理list类型
        if ($queue !== 'reserved') {
            while ($this->redis->rpoplpush($channel, $this->channel->getWaiting())) {
                ++$num;
            }
            return $num;
        }
        //处理zset类型
        while ($job = $this->redis->zPopMin($channel)) {
            $this->redis->lPush($this->channel->getWaiting(), key($job));
            ++$num;
        }
        return $num;
    }
}